import os
import pandas as pd

from tgsdata.kafka_kit import Kafka_producer
from utils.tools import to_pdtime_ms, to_pdtime_str, get_mysql

columns = ['VERSION', 'PASSTIME', 'CARSTATE', 'CARPLATE', 'PLATETYPE', 'SPEED', 'PLATECOLOR', 'LOCATIONID'
    , 'DEVICEID', 'DRIVEWAY', 'DRIVEDIR', 'CAPTUREDIR', 'CARCOLOR', 'CARBRAND', 'CARBRANDZW', 'TGSID', 'PLATECOORD'
    , 'CABCOORD', 'IMGID1', 'IMGID2', 'IMGID3']

columns_use = ['PASSTIME', 'CARPLATE', 'SPEED', 'DRIVEWAY', 'DRIVEDIR', 'CAPTUREDIR', 'PLATETYPE', 'TGSID']
producer = Kafka_producer(bootstrap_servers=['10.2.111.56:9092'], kafkatopic='passcar')


def read_data(file):
    list = []
    with open(file ,"r",encoding="utf-8") as lines:
        for line in lines.read().strip().split("\n"):
            data = []
            if line.startswith("VERSION=1.0"):
                field = line.split("\t")
                for dic in field:
                    value = dic.split("=")
                    if len(value)>0:
                        data.append(value[1])
                list.append(data)
    df = pd.DataFrame(list,columns=columns)
    return df

def file_to_kafka(file):
    print('converting file:%r(%r MB)' % (file, (os.path.getsize(file) / (1024 * 1024))))
    df_org = read_data(file)
    df = df_org[columns_use].copy()

    df['PASSTIME'] = to_pdtime_str(df['PASSTIME'].str[0:19],tz=False)
    df = df[df['PASSTIME'].notnull()]

    df['DRIVEDIR'] = df['DRIVEDIR'].astype('int32', errors='ignore')
    df['DRIVEWAY'] = df['DRIVEWAY'].astype('int32', errors='ignore')

    data_len = len(df)
    count = int(data_len / 100)
    count = count if data_len % 100 == 0 else count + 1
    for i in range(count):
        producer.send_df(df.iloc[i * 100:(i + 1) * 100])

    print(f'save {file}')


def read_file():
    dir = r'E:\EHL\CODE\python\项目\新乡\pr'
    for root, dirs, files in os.walk(dir):
        for file in files:
            dir = os.path.join(root,file)
            file_to_kafka(dir)






if __name__ == '__main__':
    print('start')
    read_file()
    # file_to_kafka(r"E:\EHL\CODE\python\项目\新乡\pr\20181022\41.111.32.20-tvc-2018-10-22-00")
    print('传输kafka完成')
